package tcpServer

/*
 *非阻塞型TCP服务器
 */

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

	"github.com/RuiGuangServer/config"
	"github.com/RuiGuangServer/global"
	"github.com/RuiGuangServer/logs"
	"github.com/RuiGuangServer/protocol"
	"github.com/RuiGuangServer/utils"
)

// TCPServer TCP服务端结构体
type TCPServer struct {
	addr               string
	listener           net.Listener
	clients            sync.Map
	stopChan           chan struct{}
	wg                 sync.WaitGroup
	shutdownCtx        context.Context
	cancelFunc         context.CancelFunc
	logger             *logs.AsyncLogger
	dataer             *logs.AsyncLogger
	uuidClients        sync.Map
	broadcastChan      chan []byte
	broadcastWG        sync.WaitGroup
	semaphore          chan struct{}
	connChan           chan net.Conn
	readTimeout        time.Duration
	writeTimeout       time.Duration
	lightLevelTicker   *time.Ticker
	lightLevelStopChan chan struct{}
	clientCount        int32 // 客户端计数器
}

// NewTCPServer 创建TCP服务端实例
func NewTCPServer(addr string) *TCPServer {
	ctx, cancel := context.WithCancel(context.Background())
	return &TCPServer{
		addr:               addr,
		stopChan:           make(chan struct{}),
		shutdownCtx:        ctx,
		cancelFunc:         cancel,
		logger:             utils.LogsStart(),
		dataer:             utils.DataStart(),
		broadcastChan:      make(chan []byte, 100),
		semaphore:          make(chan struct{}, 50),
		connChan:           make(chan net.Conn, 100),
		readTimeout:        time.Duration(config.Config.ReadTimeout) * time.Millisecond,
		writeTimeout:       time.Duration(config.Config.WriteTimeout) * time.Millisecond,
		lightLevelStopChan: make(chan struct{}),
	}
}

// Start TCP连接开始
func (s *TCPServer) Start() error {
	var err error
	s.listener, err = net.Listen("tcp", s.addr)
	if err != nil {
		s.logger.Error("Failed to start TCP server: %v", err)
		return fmt.Errorf("listen error: %v", err)
	}

	s.logger.Info("TCP server started on %s", s.addr)

	// 启动各处理goroutine - 现在只有4个固定的goroutine
	s.wg.Add(4) // acceptConnections, connectionHandler, heartbeatChecker, broadcastHandler
	go s.acceptConnections()
	go s.connectionHandler()
	go s.heartbeatChecker()
	go s.broadcastHandler()

	// 灯光级别广播单独处理
	s.wg.Add(1)
	go s.startLightLevel()

	return nil
}

func (s *TCPServer) startLightLevel() {
	defer s.wg.Done()

	s.lightLevelTicker = time.NewTicker(time.Duration(config.Config.DataInterval) * time.Millisecond)
	defer s.lightLevelTicker.Stop()

	for {
		select {
		case <-s.lightLevelTicker.C:
			if s.hasRegisteredClients() {
				s.LightLevelData()
				s.broadcastAlerts()
			} else {
				s.logger.Debug("No registered clients, skipping broadcast")
			}
		case <-s.lightLevelStopChan:
			return
		case <-s.stopChan:
			return
		}
	}
}

// hasRegisteredClients 检查是否有已注册客户端
func (s *TCPServer) hasRegisteredClients() bool {
	hasClients := false
	s.clients.Range(func(key, _ any) bool {
		if client, ok := key.(*Client); ok && client.isRegistered {
			hasClients = true
			return false
		}
		return true
	})
	return hasClients
}

// LightLevelData 广播灯光数据
func (s *TCPServer) LightLevelData() {
	// 1. 获取灯光数据
	data, ok := global.LightData.Load().(map[string]*global.Listlight)
	if !ok || len(data) == 0 {
		s.logger.Debug("No light data available")
		return
	}

	// 2. 获取当前跑道信息
	var currentRunway string
	if state, ok := global.CurrentAirPortState.Load(config.Config.AirPort); ok {
		currentRunway = state.(*global.AirPortState).RWY
	}

	// 3. 处理数据
	levelData := make([]map[string]any, 0, len(data))
	for _, v := range data {
		level, err := strconv.Atoi(v.Value)
		if err != nil {
			s.logger.Warn("Invalid light level value: %s", v.Value)
			continue
		}
		item := map[string]any{
			"lightKey":  v.Key,
			"lightName": v.Name,
			"level":     level,
			"direction": currentRunway,
		}
		levelData = append(levelData, item)
	}

	if len(levelData) == 0 {
		return
	}

	// 4. 序列化并广播
	ll := map[string]any{"lights": levelData}
	jsonData, err := json.Marshal(ll)
	if err != nil {
		s.logger.Error("Failed to marshal light data: %v", err)
		return
	}

	if err := s.AddBroadcastMsg(
		protocol.NetHeadTypeProtobuf,
		string(jsonData),
		protocol.STCLightShow,
		protocol.DCNotify,
		protocol.DCAll,
	); err != nil {
		s.logger.Error("Failed to broadcast light data: %v", err)
	}
}

func (s *TCPServer) broadcastAlerts() {
	// 1. 收集告警数据
	var alertsToProcess []*global.MessageItem
	global.MsgItemCache.Range(func(key, value any) bool {
		alertsToProcess = append(alertsToProcess, value.(*global.MessageItem))
		return true
	})
	// fmt.Printf("alerts length:%d\n", len(alertsToProcess))
	if len(alertsToProcess) == 0 {
		return
	}
	at := time.Now().Unix()
	// 2. 处理告警
	for _, item := range alertsToProcess {
		bakdata, err := json.Marshal(map[string]any{
			"warnTime":  fmt.Sprintf("%d", at),
			"warnText":  item.Msgstr,
			"warnFreq":  1,
			"warnLevel": item.WarnLevel,
		})
		if err != nil {
			s.logger.Error("序列化告警数据失败: %v", err)
			continue
		}

		if item.Kind == 0 {
			if err := s.broadcastToAllClients(bakdata); err == nil {
				global.MsgItemCache.Delete(item.Typestr)
				s.logger.Debug("已广播并删除Kind=0告警: %s", item.Typestr)
			} else {
				s.logger.Error("广播Kind=0告警失败: %v", err)
			}
		} else {
			if err := s.broadcastWithUUIDTracking(item, bakdata); err != nil {
				s.logger.Error("广播Kind=%d告警失败: %v", item.Kind, err)
			}
		}
	}
}

// 广播给所有客户端（用于Kind=0）
func (s *TCPServer) broadcastToAllClients(data []byte) error {
	return s.AddBroadcastMsg(
		protocol.NetHeadTypeProtobuf,
		string(data),
		protocol.STCLightWarn,
		protocol.DCNotify,
		protocol.DCAll,
	)
}

// 广播并记录UUID（用于Kind=1,2）
func (s *TCPServer) broadcastWithUUIDTracking(msgItem *global.MessageItem, data []byte) error {
	var broadcastErrors []error
	var sentUUIDs []string

	// 遍历所有已注册客户端
	s.clients.Range(func(key, _ any) bool {
		client := key.(*Client)
		if !client.isRegistered || client.uuid == "" {
			return true
		}
		s.logger.Info("已注册uuid:%v", client.uuid)

		// 检查是否已发送过
		if msgItem.Sended != nil && msgItem.Sended[client.uuid] {
			s.logger.Debug("告警%s已发送给客户端%s，跳过", msgItem.Typestr, client.uuid)
			return true
		}

		// 发送给特定客户端
		msg, err := protocol.ReplyMsg(
			protocol.NetHeadTypeProtobuf,
			string(data),
			protocol.STCLightWarn,
			protocol.DCNotify,
			protocol.DCSingle,
		)
		if err != nil {
			broadcastErrors = append(broadcastErrors, err)
			return true
		}

		if err := client.sendProtobufMsg(msg); err != nil {
			broadcastErrors = append(broadcastErrors, err)
			return true
		}

		sentUUIDs = append(sentUUIDs, client.uuid)
		s.logger.Debug("发送告警%s给客户端%s", msgItem.Typestr, client.uuid)
		return true
	})

	// 批量更新发送记录
	if len(sentUUIDs) > 0 {
		if item, exists := global.MsgItemCache.Load(msgItem.Typestr); exists {
			msgItem := item.(*global.MessageItem)
			fmt.Printf("%v\n", msgItem)
			if msgItem.Sended == nil {
				msgItem.Sended = make(map[string]bool)
			}
			for _, uuid := range sentUUIDs {
				msgItem.Sended[uuid] = true
			}
		}
	}

	if len(broadcastErrors) > 0 {
		return fmt.Errorf("部分客户端发送失败: %v", broadcastErrors)
	}
	return nil
}

// acceptConnections 接受连接
func (s *TCPServer) acceptConnections() {
	defer s.wg.Done()

	tcpListener := s.listener.(*net.TCPListener)

	for {
		select {
		case <-s.stopChan:
			return
		default:
			_ = tcpListener.SetDeadline(time.Now().Add(s.readTimeout))
			conn, err := tcpListener.Accept()
			if err != nil {
				if errors.Is(err, net.ErrClosed) {
					return
				}
				var netErr net.Error
				if errors.As(err, &netErr) && netErr.Timeout() {
					continue
				}
				select {
				case <-s.stopChan:
					return
				default:
					s.logger.Error("Accept error: %v", err)
				}
				continue
			}

			select {
			case s.connChan <- conn:
				atomic.AddInt32(&s.clientCount, 1)
			case <-time.After(s.writeTimeout):
				conn.Close()
				s.logger.Warn("Connection channel full, closing connection")
			case <-s.stopChan:
				conn.Close()
				return
			}
		}
	}
}

// connectionHandler TCP连接处理
func (s *TCPServer) connectionHandler() {
	defer s.wg.Done()

	for {
		select {
		case conn, ok := <-s.connChan:
			if !ok {
				return
			}
			s.handleNewConnection(conn)
		case <-s.stopChan:
			return
		}
	}
}

// handleNewConnection 处理新连接
func (s *TCPServer) handleNewConnection(conn net.Conn) {
	if tcpConn, ok := conn.(*net.TCPConn); ok {
		_ = tcpConn.SetNoDelay(true)
		_ = tcpConn.SetKeepAlive(true)
		_ = tcpConn.SetKeepAlivePeriod(time.Duration(config.Config.Heartbeat) * time.Millisecond)
	}

	client := NewClient(conn, s)
	s.clients.Store(client, struct{}{})

	// 启动客户端处理goroutine，但不使用WaitGroup跟踪
	go func() {
		client.handleConnection()
		// 连接处理完成后移除客户端
		s.RemoveClient(client)
	}()
}

// RegisterUUID 在服务器注册UUID
func (s *TCPServer) RegisterUUID(uuid string, client *Client) {
	s.uuidClients.Store(uuid, client)
	s.logger.Debug("UUID registered: %s", uuid)
}

// IsUUIDRegistered 检查UUID是否已注册
func (s *TCPServer) IsUUIDRegistered(uuid string) bool {
	_, exists := s.uuidClients.Load(uuid)
	return exists
}

// UnregisterUUID 移除UUID注册
func (s *TCPServer) UnregisterUUID(uuid string) {
	s.uuidClients.Delete(uuid)
	s.logger.Debug("UUID unregistered: %s", uuid)
}

// RemoveClient 移除客户端连接
func (s *TCPServer) RemoveClient(client *Client) {
	s.clients.Delete(client)
	atomic.AddInt32(&s.clientCount, -1)
	s.logger.Debug("Client %s removed", client.conn.RemoteAddr())
}

// GetClientCount 获取当前客户端数量
func (s *TCPServer) GetClientCount() int {
	return int(atomic.LoadInt32(&s.clientCount))
}

// Stop 停止TCP服务端运行
func (s *TCPServer) Stop() {
	s.logger.Info("Shutting down TCP server...")

	close(s.stopChan)
	s.cancelFunc()

	if s.listener != nil {
		s.listener.Close()
	}

	// 关闭所有客户端连接
	s.clients.Range(func(key, value any) bool {
		if client, ok := key.(*Client); ok {
			client.Close()
		}
		return true
	})

	s.broadcastWG.Wait()
	s.wg.Wait()

	s.dataer.Close()
	s.logger.Info("TCP server stopped, client count: %d", s.GetClientCount())
	s.logger.Close()
}

// heartbeatChecker 主动心跳开启定时监控
func (s *TCPServer) heartbeatChecker() {
	defer s.wg.Done()

	ticker := time.NewTicker(time.Duration(config.Config.Heartbeat) * time.Millisecond)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			s.checkHeartbeats()
		case <-s.stopChan:
			return
		}
	}
}

// checkHeartbeats 主动心跳检查
func (s *TCPServer) checkHeartbeats() {
	timeout := time.Duration(config.Config.Heartbeat*3) * time.Millisecond
	unregisteredTimeout := 30 * time.Second

	s.clients.Range(func(key, value any) bool {
		client := key.(*Client)
		lastActive := time.Since(client.lastActive)

		if client.isRegistered {
			if lastActive > timeout {
				s.logger.Warn("Registered client %s heartbeat timeout, closing connection",
					client.conn.RemoteAddr())
				client.Close()
				s.clients.Delete(client)
			} else if err := client.activeSendHeartbeat(); err != nil {
				s.logger.Warn("Failed to send heartbeat to %s: %v",
					client.conn.RemoteAddr(), err)
				client.Close()
				s.clients.Delete(client)
			}
		} else if lastActive > unregisteredTimeout {
			s.logger.Warn("Unregistered client %s connection timeout, closing",
				client.conn.RemoteAddr())
			client.Close()
			s.clients.Delete(client)
		}
		return true
	})
}

// broadcastToRegistered 向所有已注册客户端广播消息
func (s *TCPServer) broadcastToRegistered(message []byte) error {
	select {
	case s.broadcastChan <- message:
		return nil
	case <-time.After(time.Second):
		return fmt.Errorf("broadcast queue is full")
	case <-s.shutdownCtx.Done():
		return fmt.Errorf("server is shutting down")
	}
}

// broadcastHandler 处理广播消息并发送
func (s *TCPServer) broadcastHandler() {
	defer s.wg.Done()

	for {
		select {
		case msg, ok := <-s.broadcastChan:
			if !ok {
				return
			}
			s.broadcastWG.Add(1)
			go s.processBroadcastMessage(msg)
		case <-s.stopChan:
			return
		}
	}
}

// processBroadcastMessage 处理单个广播消息
func (s *TCPServer) processBroadcastMessage(msg []byte) {
	defer s.broadcastWG.Done()

	s.semaphore <- struct{}{}
	defer func() { <-s.semaphore }()

	var errs []error
	s.clients.Range(func(key, _ any) bool {
		client := key.(*Client)
		if client.isRegistered {
			if err := client.sendProtobufMsg(msg); err != nil {
				errs = append(errs, fmt.Errorf("client %s: %v", client.conn.RemoteAddr(), err))
				client.Close()
				s.clients.Delete(client)
			}
		}
		return true
	})

	if len(errs) > 0 {
		s.logger.Warn("Broadcast completed with %d errors", len(errs))
	}
}

// AddBroadcastMsg 添加广播消息
func (s *TCPServer) AddBroadcastMsg(headType uint8, data string, protoType int32, protoProto int32, protoExt int32) error {
	if !s.hasRegisteredClients() {
		s.logger.Debug("No registered clients, skipping broadcast message")
		return nil
	}

	msg, err := protocol.ReplyMsg(headType, data, protoType, protoProto, protoExt)
	if err != nil {
		return fmt.Errorf("create message failed: %v", err)
	}
	now := time.Now()
	nt := now.Format("15:04:05.000")
	fmt.Printf("%v 发送广播：%x\n", nt, msg)
	s.logger.Debug("Broadcasting message type: 0x%x", protoType)
	return s.broadcastToRegistered(msg)
}
